package com.atguigu.function;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.GmallConfig;
import com.atguigu.util.DruidPhoenixUtil;
import com.atguigu.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

/**
 * @author yhm
 * @create 2022-11-18 9:33
 */
public class MyBroadcastFunction extends BroadcastProcessFunction<String, String, JSONObject> {

    // key:SourceTable   value:jsonString
    public HashMap<String, String> configMap = new HashMap<String, String>();

    public MapStateDescriptor<String, TableProcess> tableProcessStateDes;

    public MyBroadcastFunction(MapStateDescriptor<String, TableProcess> tableProcessStateDes) {
        this.tableProcessStateDes = tableProcessStateDes;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 在flink程序运行之前的操作
        Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config", "root", "123456");

        PreparedStatement ps = connection.prepareStatement("select * from table_process");

        ResultSet resultSet = ps.executeQuery();
        ResultSetMetaData metaData = resultSet.getMetaData();
        while (resultSet.next()) {
            JSONObject jsonObject = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                jsonObject.put(metaData.getColumnName(i), resultSet.getString(i));
            }
            configMap.put(jsonObject.getString("source_table"), jsonObject.toString());

            // 同时检查表格是否存在
            checkTableExists(jsonObject.getString("sink_table"),jsonObject.getString("sink_columns"),jsonObject.getString("sink_pk"),jsonObject.getString("sink_extend"));

        }

        ps.close();
        connection.close();

    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        // TODO 1 解析接受的数据
        JSONObject jsonObject = JSON.parseObject(value);
        // 获取状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableProcessStateDes);

        // 获取当前数据的类型
        String op = jsonObject.getString("op");
        if ("d".equals(op)) {
            try {
                // 此处为维度表删除数据
                // 只需要将当前维度表去除即可  之后不再处理
                String deleteKey = jsonObject.getJSONObject("before").getString("source_table");
                configMap.remove(deleteKey);
                broadcastState.remove(deleteKey);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        String after = jsonObject.getString("after");

        TableProcess tableProcess = JSON.parseObject(after, TableProcess.class);

        if ("dim".equals(tableProcess.getSinkType())) {
            // TODO 2 判断表格是否已经在phoenix创建  如果不存在创建出来
            checkTableExists(tableProcess.getSinkTable(), tableProcess.getSinkColumns(), tableProcess.getSinkPk(), tableProcess.getSinkExtend());

            // TODO 3 将数据写入到状态  进行广播
            broadcastState.put(tableProcess.getSourceTable(), tableProcess);
        }
    }

    // 判断表格是否已经在phoenix创建  如果不存在创建出来
    private void checkTableExists(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) throws SQLException {
        // 获取phoenix连接
        DruidPooledConnection phoenixConn = DruidPhoenixUtil.getPhoenixConn();
        // 创建
        String sql = createSql(sinkTable, sinkColumns, sinkPk, sinkExtend);
        // 执行sql
        PhoenixUtil.executeSql(phoenixConn, sql);
        // 关闭连接
        phoenixConn.close();
    }

    private static String createSql(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        // create table if not exists sinkTable(column varchar primary key , column varchar...) sinkExtend
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append("(");

        if (sinkPk == null) {
            sinkPk = "id";
        }

        if (sinkExtend == null) {
            sinkExtend = "";
        }

        String[] columns = sinkColumns.split(",");
        for (int i = 0; i < columns.length; i++) {
            if (sinkPk.equals(columns[i])){
                stringBuilder.append(columns[i])
                        .append(" varchar primary key ");
            }else {
                stringBuilder.append(columns[i])
                        .append(" varchar ");
            }

            if (i < columns.length - 1){
                stringBuilder.append(",");
            }
        }
        stringBuilder.append(")")
                .append(sinkExtend);


        return stringBuilder.toString();
    }

    @Override
    public void processElement(String value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // TODO 1 获取配置流的状态数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableProcessStateDes);

        // TODO 2 判断是否为维度表
        JSONObject maxwellJson = JSON.parseObject(value);
        // 获取source_table
        String database = maxwellJson.getString("database");
        if ("gmall".equals(database)){
            String table = maxwellJson.getString("table");
            System.out.println(table);
            if (broadcastState.contains(table) || configMap.containsKey(table)) {
                // TODO 3 添加过滤字段 只保留需要写出到phoenix的字段和sink_table
                // 找到对应的tableProcess
                TableProcess tableProcess = broadcastState.get(table);
                if (tableProcess == null){
                    tableProcess = JSON.parseObject(configMap.get(table),TableProcess.class);
                }
                JSONObject data = maxwellJson.getJSONObject("data");
                System.out.println(data);
                String sinkColumns = tableProcess.getSinkColumns();
                // 过滤字段
                filterSinkColumns(data,sinkColumns);

                data.put("sink_table",tableProcess.getSinkTable());

                out.collect(data);
            }
        }

    }

    // 过滤json数据  只保留sinkColumns中包含的字段
    private static void filterSinkColumns(JSONObject data, String sinkColumns) {

        // 如果sinkColumns   -> activity_user_info_id,id,info_info
        // data   -> user_info_id
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        List<String> strings = Arrays.asList(sinkColumns.split(","));

        // 使用迭代器删除
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()) {
            String key = iterator.next().getKey();
            if (!strings.contains(key)){
                iterator.remove();
            }
        }
//        entries.removeIf(entry -> !strings.contains(entry.getKey()));

    }


}
